package com.lagou.client.boot;

import com.lagou.client.RPCConsumer;
import com.lagou.service.IUserServer;
import com.lagou.util.ZkUtil;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang3.StringUtils;

import java.util.*;

public class ConsumerBoot {

    // 记录每个服务的代理类
    private static Map<String, IUserServer> iUserServerHashMap = new HashMap<>();

    // 记录当前服务列表
    private static List<String> serverAddressList = new ArrayList<>();

    private static ZkClient zkClient = new ZkClient(ZkUtil.ZK_ADDRESS);

    // 用于中间存储，根据服务端返回的花费时间costTime，对服务列表排序
    private static SortedMap<Integer, String> sortMap = new TreeMap<>();

    // 服务端返回的花费时间costTime，是否相同
    private static boolean isEqual = false;

    public static void main(String[] args) throws Exception {

        // 监听zk子列表变化
        subZkChildChanges();
//
        serverAddressList = getServerList();
        for (String address : serverAddressList) {
            // 监听子列表的值变化
            subDataChange(address);
            IUserServer server = (IUserServer) RPCConsumer.createProxy(IUserServer.class, address);
            iUserServerHashMap.put(address, server);
        }

        int serverIndex = 0;
        while (true) {
            String serverAddress = "";
            try {
                if (isEqual) {
                    serverIndex = (int) (Math.random() * serverAddressList.size());
                }
                serverAddress = serverAddressList.get(serverIndex);
                IUserServer server = iUserServerHashMap.get(serverAddress);
                String s = server.sayHello("send client heartbeat......");
                System.out.println(serverAddressList.get(0) + " 服务,返回结果：" + s);
                Thread.sleep((long) (Math.random() * 3000));
            } catch (Exception e) {
                if (serverAddressList.size() == 0) {
                    System.out.println("当前服务列表不存在，重连中");
                } else {
                    // 负载均衡 获取下一个服务
                    if (StringUtils.isBlank(serverAddress)) {
                        return;
                    }
                    serverAddressList = getServerList();
                    System.out.println("服务端 " + serverAddressList.get(serverIndex) + "连接失败，重连中");
                }
                Thread.sleep(5000L);
            }
        }
    }

    /**
     * 获取服务器地址列表
     *
     * @return
     */
    private static List<String> getServerIpList() {
        List<String> children = null;
        if (zkClient.exists(ZkUtil.PATH)) {
            children = zkClient.getChildren(ZkUtil.PATH);
        }
        return children;
    }

    /**
     * 监听/netty_rpc下子列表的变更
     */
    private static void subZkChildChanges() {
        zkClient.subscribeChildChanges(ZkUtil.PATH, (parentPath, currentChildren) -> {
            serverAddressList = currentChildren;

            for (String child : currentChildren) {
                // 说明有服务端上线
                if (iUserServerHashMap.getOrDefault(child, null) == null) {
                    System.out.println("zk watcher====>" + child + "：服务新上线");
                    IUserServer server = (IUserServer) RPCConsumer.createProxy(IUserServer.class, child);
                    iUserServerHashMap.put(child, server);
                }
            }

            for (Map.Entry<String, IUserServer> entry : iUserServerHashMap.entrySet()) {
                if (!currentChildren.contains(entry.getKey())) {
                    iUserServerHashMap.remove(entry.getKey());
                    System.out.println("zk watcher====>" + entry.getKey() + "：服务断开");
                }
            }
        });
    }

    /**
     * 监听/netty_rpc下每个子列表的数据变化
     *
     * @param address
     */
    private static void subDataChange(String address) {
        zkClient.subscribeDataChanges(ZkUtil.PATH + "/" + address, new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                if (serverAddressList == null || serverAddressList.size() < 1) {
                    return;
                }

                getServerList();
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {

            }
        });
    }

    /**
     * 获取ServerList
     *
     * @return
     */
    private static List<String> getServerList() {
        List<String> serverIpList = getServerIpList();
        sortMap.clear();
        for (String serverIp : serverIpList) {
            Object readData = zkClient.readData(ZkUtil.PATH + "/" + serverIp);
            String[] split = readData.toString().split("_");
            sortMap.put(Integer.valueOf(split[1]), serverIp);
        }

        if (sortMap.size() > 1) {
            serverIpList.clear();
            for (Map.Entry<Integer, String> entry : sortMap.entrySet()) {
                serverIpList.add(entry.getValue());
            }
        } else {
            isEqual = true;
        }
        return serverIpList;
    }
}
